package com.flink.hbase.windowingoptimization.monitoring;

import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.metrics.Counter;
import org.apache.flink.metrics.Gauge;
import org.apache.flink.metrics.Histogram;
import org.apache.flink.metrics.Meter;
import org.apache.flink.metrics.MetricGroup;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;

/**
 * 窗口监控工具和指标收集系统
 * 
 * 监控指标：
 * 1. 窗口处理延迟
 * 2. 窗口大小分布
 * 3. 数据倾斜检测
 * 4. 反压监控
 * 5. 内存使用监控
 * 6. 吞吐量监控
 * 
 * 告警功能：
 * - 延迟超阈值告警
 * - 数据倾斜告警
 * - 反压告警
 * - 内存不足告警
 * 
 * @author Flink Developer
 */
public class WindowingMetricsCollector {
    
    private static final Logger logger = LoggerFactory.getLogger(WindowingMetricsCollector.class);
    
    /**
     * 窗口监控事件
     */
    public static class WindowingEvent {
        public String key;
        public Object value;
        public long timestamp;
        public String eventType;
        public long eventSize;
        public Map<String, Object> metadata;
        
        public WindowingEvent() {
            this.metadata = new ConcurrentHashMap<>();
        }
        
        public WindowingEvent(String key, Object value, long timestamp, 
                            String eventType, long eventSize) {
            this.key = key;
            this.value = value;
            this.timestamp = timestamp;
            this.eventType = eventType;
            this.eventSize = eventSize;
            this.metadata = new ConcurrentHashMap<>();
        }
        
        @Override
        public String toString() {
            return String.format("WindowingEvent{key='%s', eventType='%s', timestamp=%d, eventSize=%d}",
                    key, eventType, timestamp, eventSize);
        }
    }
    
    /**
     * 窗口指标
     */
    public static class WindowMetrics {
        public String windowKey;
        public long windowStart;
        public long windowEnd;
        public long eventCount;
        public long totalSize;
        public long processingStartTime;
        public long processingEndTime;
        public long processingLatency;
        public double throughput;
        public long memoryUsage;
        public boolean hasBackpressure;
        public double skewFactor;
        
        public WindowMetrics() {
            this.processingStartTime = System.currentTimeMillis();
        }
        
        public void finishProcessing() {
            this.processingEndTime = System.currentTimeMillis();
            this.processingLatency = processingEndTime - processingStartTime;
            this.throughput = processingLatency > 0 ? 
                (double) eventCount / (processingLatency / 1000.0) : 0;
        }
        
        @Override
        public String toString() {
            return String.format("WindowMetrics{windowKey='%s', eventCount=%d, " +
                    "processingLatency=%d ms, throughput=%.2f events/s, " +
                    "memoryUsage=%d bytes, hasBackpressure=%b, skewFactor=%.2f}",
                    windowKey, eventCount, processingLatency, throughput, 
                    memoryUsage, hasBackpressure, skewFactor);
        }
    }
    
    /**
     * 指标收集器
     */
    public static class MetricsCollector extends ProcessFunction<WindowingEvent, WindowingEvent> {
        
        // Flink内置指标
        private transient Counter eventCounter;
        private transient Counter errorCounter;
        private transient Meter eventRate;
        private transient Histogram eventSizeHistogram;
        private transient Gauge<Long> memoryUsageGauge;
        private transient Gauge<Double> backpressureGauge;
        
        // 自定义指标
        private final Map<String, AtomicLong> keyCounters = new ConcurrentHashMap<>();
        private final AtomicLong totalEvents = new AtomicLong(0);
        private final AtomicReference<Double> currentBackpressure = new AtomicReference<>(0.0);
        
        private long lastMetricsTime = System.currentTimeMillis();
        
        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            
            MetricGroup metricGroup = getRuntimeContext().getMetricGroup()
                .addGroup("windowing")
                .addGroup("metrics");
            
            // 初始化Flink指标
            eventCounter = metricGroup.counter("events_processed");
            errorCounter = metricGroup.counter("errors");
            eventRate = metricGroup.meter("event_rate");
            eventSizeHistogram = metricGroup.histogram("event_size");
            
            // 内存使用监控
            memoryUsageGauge = metricGroup.gauge("memory_usage", () -> {
                Runtime runtime = Runtime.getRuntime();
                return runtime.totalMemory() - runtime.freeMemory();
            });
            
            // 反压监控
            backpressureGauge = metricGroup.gauge("backpressure_ratio", () -> 
                currentBackpressure.get());
        }
        
        @Override
        public void processElement(WindowingEvent event, Context context, 
                                 Collector<WindowingEvent> collector) throws Exception {
            
            try {
                // 更新指标
                eventCounter.inc();
                eventRate.markEvent();
                eventSizeHistogram.update(event.eventSize);
                
                // 更新自定义指标
                keyCounters.computeIfAbsent(event.key, k -> new AtomicLong(0)).incrementAndGet();
                totalEvents.incrementAndGet();
                
                // 检测数据倾斜
                detectDataSkew(event);
                
                // 检测反压
                detectBackpressure(context);
                
                // 定期输出监控报告
                long currentTime = System.currentTimeMillis();
                if (currentTime - lastMetricsTime > 30000) { // 30秒
                    generateMetricsReport();
                    lastMetricsTime = currentTime;
                }
                
                collector.collect(event);
                
            } catch (Exception e) {
                errorCounter.inc();
                logger.error("处理事件时发生错误: {}", event, e);
                throw e;
            }
        }
        
        private void detectDataSkew(WindowingEvent event) {
            long total = totalEvents.get();
            if (total > 1000) { // 至少1000个事件后开始检测
                long keyCount = keyCounters.get(event.key).get();
                double percentage = (double) keyCount / total;
                
                if (percentage > 0.1) { // 超过10%认为倾斜
                    logger.warn("检测到数据倾斜: key={}, 占比={:.2f}%", 
                        event.key, percentage * 100);
                }
            }
        }
        
        private void detectBackpressure(Context context) {
            // 简化的反压检测逻辑
            long currentTime = System.currentTimeMillis();
            long processingTime = context.timestamp();
            
            if (processingTime > 0) {
                double backpressureRatio = Math.min(1.0, 
                    (double) (currentTime - processingTime) / 1000.0);
                currentBackpressure.set(backpressureRatio);
                
                if (backpressureRatio > 0.5) {
                    logger.warn("检测到反压: 比例={:.2f}", backpressureRatio);
                }
            }
        }
        
        private void generateMetricsReport() {
            logger.info("=== 窗口监控报告 ===");
            logger.info("总处理事件数: {}", totalEvents.get());
            logger.info("错误事件数: {}", errorCounter.getCount());
            logger.info("当前事件处理速率: {:.2f} events/s", eventRate.getRate());
            
            // 内存使用情况
            long memoryUsage = memoryUsageGauge.getValue();
            logger.info("内存使用: {} MB", memoryUsage / 1024 / 1024);
            
            // 反压情况
            double backpressure = backpressureGauge.getValue();
            logger.info("反压比例: {:.2f}", backpressure);
            
            // Top 5 热点key
            keyCounters.entrySet().stream()
                .sorted(Map.Entry.<String, AtomicLong>comparingByValue(
                    (a, b) -> Long.compare(b.get(), a.get())))
                .limit(5)
                .forEach(entry -> 
                    logger.info("热点key: {} ({}次)", entry.getKey(), entry.getValue().get()));
            
            logger.info("===================");
        }
    }
    
    /**
     * 窗口性能监控函数
     */
    public static class WindowPerformanceMonitor<K> 
            extends ProcessWindowFunction<WindowingEvent, WindowMetrics, K, TimeWindow> {
        
        private transient Counter windowCounter;
        private transient Histogram windowSizeHistogram;
        private transient Histogram processingLatencyHistogram;
        private transient Gauge<Long> activeWindowsGauge;
        
        private final AtomicLong activeWindows = new AtomicLong(0);
        
        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            
            MetricGroup metricGroup = getRuntimeContext().getMetricGroup()
                .addGroup("windowing")
                .addGroup("window_performance");
            
            windowCounter = metricGroup.counter("windows_processed");
            windowSizeHistogram = metricGroup.histogram("window_size");
            processingLatencyHistogram = metricGroup.histogram("processing_latency");
            activeWindowsGauge = metricGroup.gauge("active_windows", activeWindows::get);
        }
        
        @Override
        public void process(K key, Context context, Iterable<WindowingEvent> elements, 
                          Collector<WindowMetrics> out) throws Exception {
            
            activeWindows.incrementAndGet();
            
            WindowMetrics metrics = new WindowMetrics();
            metrics.windowKey = key.toString();
            metrics.windowStart = context.window().getStart();
            metrics.windowEnd = context.window().getEnd();
            
            // 计算窗口统计
            long eventCount = 0;
            long totalSize = 0;
            
            for (WindowingEvent event : elements) {
                eventCount++;
                totalSize += event.eventSize;
            }
            
            metrics.eventCount = eventCount;
            metrics.totalSize = totalSize;
            metrics.finishProcessing();
            
            // 检查窗口延迟
            long windowDelay = System.currentTimeMillis() - metrics.windowEnd;
            if (windowDelay > 10000) { // 超过10秒
                logger.warn("窗口处理延迟过高: key={}, 延迟={}ms", key, windowDelay);
            }
            
            // 检查窗口大小异常
            if (eventCount > 100000) { // 超过10万事件
                logger.warn("窗口大小异常: key={}, 事件数={}", key, eventCount);
            }
            
            // 估算内存使用
            metrics.memoryUsage = estimateMemoryUsage(eventCount, totalSize);
            
            // 计算数据倾斜因子
            metrics.skewFactor = calculateSkewFactor(eventCount);
            
            // 更新指标
            windowCounter.inc();
            windowSizeHistogram.update(eventCount);
            processingLatencyHistogram.update(metrics.processingLatency);
            
            activeWindows.decrementAndGet();
            
            out.collect(metrics);
        }
        
        private long estimateMemoryUsage(long eventCount, long totalSize) {
            // 简化的内存使用估算
            return eventCount * 64 + totalSize; // 假设每个事件64字节开销
        }
        
        private double calculateSkewFactor(long eventCount) {
            // 简化的倾斜因子计算
            // 在实际应用中，需要基于历史数据和统计分析
            return eventCount > 1000 ? Math.log(eventCount) / 10.0 : 0.0;
        }
    }
    
    /**
     * 告警管理器
     */
    public static class AlertManager {
        
        private static final Logger alertLogger = LoggerFactory.getLogger("ALERT");
        
        private final Map<String, Long> alertHistory = new ConcurrentHashMap<>();
        private final long alertCooldown = 300000; // 5分钟冷却时间
        
        public void checkAndSendAlert(String alertType, String message, Object... params) {
            String alertKey = alertType + "_" + String.format(message, params);
            long currentTime = System.currentTimeMillis();
            
            Long lastAlertTime = alertHistory.get(alertKey);
            if (lastAlertTime == null || currentTime - lastAlertTime > alertCooldown) {
                sendAlert(alertType, message, params);
                alertHistory.put(alertKey, currentTime);
            }
        }
        
        private void sendAlert(String alertType, String message, Object... params) {
            String formattedMessage = String.format(message, params);
            alertLogger.error("[{}] {}", alertType, formattedMessage);
            
            // 这里可以集成其他告警系统
            // 如钉钉、邮件、短信等
            
            switch (alertType) {
                case "LATENCY":
                    handleLatencyAlert(formattedMessage);
                    break;
                case "BACKPRESSURE":
                    handleBackpressureAlert(formattedMessage);
                    break;
                case "MEMORY":
                    handleMemoryAlert(formattedMessage);
                    break;
                case "SKEW":
                    handleSkewAlert(formattedMessage);
                    break;
                default:
                    logger.warn("未知告警类型: {}", alertType);
            }
        }
        
        private void handleLatencyAlert(String message) {
            // 处理延迟告警
            logger.error("延迟告警: {}", message);
        }
        
        private void handleBackpressureAlert(String message) {
            // 处理反压告警
            logger.error("反压告警: {}", message);
        }
        
        private void handleMemoryAlert(String message) {
            // 处理内存告警
            logger.error("内存告警: {}", message);
        }
        
        private void handleSkewAlert(String message) {
            // 处理数据倾斜告警
            logger.error("数据倾斜告警: {}", message);
        }
    }
    
    /**
     * 监控数据聚合器
     */
    public static class MonitoringAggregator extends RichMapFunction<WindowMetrics, String> {
        
        private transient AlertManager alertManager;
        private final Map<String, WindowMetrics> metricsHistory = new ConcurrentHashMap<>();
        
        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            alertManager = new AlertManager();
        }
        
        @Override
        public String map(WindowMetrics metrics) throws Exception {
            // 存储历史指标
            metricsHistory.put(metrics.windowKey, metrics);
            
            // 检查各种告警条件
            checkLatencyAlert(metrics);
            checkBackpressureAlert(metrics);
            checkMemoryAlert(metrics);
            checkSkewAlert(metrics);
            
            // 生成监控报告
            return generateMonitoringReport(metrics);
        }
        
        private void checkLatencyAlert(WindowMetrics metrics) {
            if (metrics.processingLatency > 5000) { // 超过5秒
                alertManager.checkAndSendAlert("LATENCY", 
                    "窗口处理延迟过高: key=%s, 延迟=%d ms", 
                    metrics.windowKey, metrics.processingLatency);
            }
        }
        
        private void checkBackpressureAlert(WindowMetrics metrics) {
            if (metrics.hasBackpressure) {
                alertManager.checkAndSendAlert("BACKPRESSURE", 
                    "检测到反压: key=%s, 事件数=%d", 
                    metrics.windowKey, metrics.eventCount);
            }
        }
        
        private void checkMemoryAlert(WindowMetrics metrics) {
            if (metrics.memoryUsage > 100 * 1024 * 1024) { // 超过100MB
                alertManager.checkAndSendAlert("MEMORY", 
                    "内存使用过高: key=%s, 使用量=%d MB", 
                    metrics.windowKey, metrics.memoryUsage / 1024 / 1024);
            }
        }
        
        private void checkSkewAlert(WindowMetrics metrics) {
            if (metrics.skewFactor > 2.0) {
                alertManager.checkAndSendAlert("SKEW", 
                    "数据倾斜严重: key=%s, 倾斜因子=%.2f", 
                    metrics.windowKey, metrics.skewFactor);
            }
        }
        
        private String generateMonitoringReport(WindowMetrics metrics) {
            StringBuilder report = new StringBuilder();
            report.append("窗口监控报告:\n");
            report.append(String.format("  窗口: %s\n", metrics.windowKey));
            report.append(String.format("  时间范围: %d - %d\n", 
                metrics.windowStart, metrics.windowEnd));
            report.append(String.format("  事件数: %d\n", metrics.eventCount));
            report.append(String.format("  处理延迟: %d ms\n", metrics.processingLatency));
            report.append(String.format("  吞吐量: %.2f events/s\n", metrics.throughput));
            report.append(String.format("  内存使用: %d bytes\n", metrics.memoryUsage));
            report.append(String.format("  反压状态: %s\n", 
                metrics.hasBackpressure ? "是" : "否"));
            report.append(String.format("  倾斜因子: %.2f\n", metrics.skewFactor));
            
            return report.toString();
        }
    }
    
    /**
     * 性能基准测试
     */
    public static class PerformanceBenchmark {
        
        private final Map<String, Long> benchmarkResults = new ConcurrentHashMap<>();
        
        public void runBenchmark(String testName, Runnable test) {
            logger.info("开始基准测试: {}", testName);
            
            long startTime = System.currentTimeMillis();
            long startMemory = getUsedMemory();
            
            try {
                test.run();
            } catch (Exception e) {
                logger.error("基准测试失败: {}", testName, e);
                return;
            }
            
            long endTime = System.currentTimeMillis();
            long endMemory = getUsedMemory();
            
            long duration = endTime - startTime;
            long memoryUsed = endMemory - startMemory;
            
            benchmarkResults.put(testName + "_duration", duration);
            benchmarkResults.put(testName + "_memory", memoryUsed);
            
            logger.info("基准测试完成: {}, 耗时: {} ms, 内存使用: {} bytes", 
                testName, duration, memoryUsed);
        }
        
        private long getUsedMemory() {
            Runtime runtime = Runtime.getRuntime();
            return runtime.totalMemory() - runtime.freeMemory();
        }
        
        public void printBenchmarkResults() {
            logger.info("=== 基准测试结果 ===");
            benchmarkResults.forEach((key, value) -> 
                logger.info("{}: {}", key, value));
            logger.info("==================");
        }
    }
    
    /**
     * 监控配置
     */
    public static class MonitoringConfig {
        public long metricsReportInterval = 30000; // 30秒
        public long alertCooldown = 300000; // 5分钟
        public long latencyThreshold = 5000; // 5秒
        public long memoryThreshold = 100 * 1024 * 1024; // 100MB
        public double skewThreshold = 2.0;
        public boolean enableDetailedLogging = false;
        
        public static MonitoringConfig defaultConfig() {
            return new MonitoringConfig();
        }
        
        public static MonitoringConfig highSensitivityConfig() {
            MonitoringConfig config = new MonitoringConfig();
            config.metricsReportInterval = 10000; // 10秒
            config.alertCooldown = 60000; // 1分钟
            config.latencyThreshold = 1000; // 1秒
            config.memoryThreshold = 50 * 1024 * 1024; // 50MB
            config.skewThreshold = 1.5;
            config.enableDetailedLogging = true;
            return config;
        }
    }
} 